package com.gg.communication.communication;

import com.gg.communication.event.ReceiveListener;
import com.gg.communication.event.check.CheckDecorator;
import com.gg.communication.exception.CheckException;
import com.gg.communication.exception.QueueWaitingException;
import com.gg.communication.model.MessageTask;
import com.gg.communication.model.MessageType;
import com.gg.communication.model.TaskState;

import java.io.IOException;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author sakura
 * @since 2025-06-16
 */
public abstract class BaseCommunication {

    private String host;

    private int port;

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    /**
     * 排队超时
     */
    public long queueTimeout = 3000;

    /**
     * 重试次数
     */
    private int retryCount = 5;

    public int getRetryCount() {
        return retryCount;
    }

    public void setRetryCount(int retryCount) {
        this.retryCount = retryCount;
    }

    /**
     * 等待消息超时
     */

    public long responseTimeout = 5000;

    public long getResponseTimeout() {
        return responseTimeout;
    }

    public void setResponseTimeout(long responseTimeout) {
        this.responseTimeout = responseTimeout;
    }

    /**
     * 正在处理的任务
     */
    protected MessageTask task;
    private final Lock lock = new ReentrantLock();

    private final Queue<MessageTask> messageQueue = new LinkedList<>();
    private final Queue<MessageTask> messageQueue2 = new LinkedList<>();

    private final Condition notEmpty = lock.newCondition();

    private ExecutorService executor;

    public void setWait(boolean wait) {
        isWait = wait;
    }

    private boolean isWait = true;

    private CheckDecorator reveCheck;


    public ReceiveListener receiveListener;

    private static BaseCommunication INSTANCE;

    protected boolean isRun = false;

    protected static BaseCommunication getInstance(BaseCommunication instance) {
        synchronized (BaseCommunication.class) {
            if (INSTANCE == null) {
                INSTANCE = instance;
            }
        }
        return INSTANCE;
    }

    public static BaseCommunication getInstance() {
        return INSTANCE;
    }


    protected BaseCommunication() {
    }

    public void setReveCheck(CheckDecorator reveCheck) {
        this.reveCheck = reveCheck;
    }

    public void setReceiveListener(ReceiveListener receiveListener) {
        this.receiveListener = receiveListener;
    }

    /**
     * 初始化
     *
     * @throws SocketException      如果无法创建socket
     * @throws UnknownHostException 如果主机无法解析
     */
    public void start() throws IOException {
        if (!isRun) {
            isRun = true;
            executor = Executors.newFixedThreadPool(2);
            startCommunication();
            // 启动发送线程
            executor.submit(this::processQueue);
            executor.submit(this::received);
        }
    }

    protected abstract void startCommunication() throws IOException;

    int sendNumber;


    boolean waitMessage = false;

    /**
     * 处理消息队列的线程方法
     */
    protected void processQueue() {
        while (isRun) {
            System.out.println("processQueue");
            try {
                if (task == null || task.isReceived()) {
                    task = take();
                    sendNumber = 0;
                }

                //// 检查是否已经超时
//                System.out.println("ttttttttt");
//                System.out.println(task.isSendTimedOut());
                if (task.isSendTimedOut()) {
//                    task.future.completeExceptionally(new TimeoutException("Message timed out in queue before sending"));
//                    task = null;
//                    continue;
                    task.enqueueTime = System.currentTimeMillis();
                    throw new CheckException();
                }

                if (!waitMessage) {
                    sendNumber++;
                    doCommunicationSend(task.message);
                    task.markAsSent();
                }
                if (!isWait) {
                    task.future.complete(new byte[0]);
                    task.receive();
                    continue;
                }
                //// 准备接收响应

                byte[] responseData = null;
                try {
                    task.receiveFuture = new CompletableFuture<>();
                    responseData = task.receiveFuture.get(task.getSendTimeout(), TimeUnit.MILLISECONDS);
                } catch (TimeoutException e) {
                    //task.future.completeExceptionally(e);
                    throw new SocketTimeoutException();
                }


                if (reveCheck == null) {
                    task.future.complete(responseData);
                    task.receive();
                } else {

                    if (sendNumber < retryCount && task.getState().isWaitingTimeout()) {
                        if (reveCheck.check(responseData)) {
                            //responseData = reveCheck.recvMsg;
                            task.future.complete(responseData);
                            task.receive();
                            //continue;
                            waitMessage = false;
                        } else {
                            waitMessage = true;
                        }
                    } else {
                        task.future.completeExceptionally(new CheckException());
                        task.receive();
                    }
                }


            } catch (QueueWaitingException e) {
                throw new QueueWaitingException(e.getMessage());
            } catch (Exception e) {
                waitMessage = false;
                e.printStackTrace();
                if (sendNumber >= retryCount || !task.getState().isWaitingTimeout()) {
                    task.receive();
                    task.future.completeExceptionally(new SocketTimeoutException("Response timed out after " + responseTimeout * sendNumber + "ms"));
                }
            }
        }

    }


    public void doCommunicationSend(byte[] msg) throws IOException {
        doCommunicationSend(msg, host, port);
    }

    public abstract void doCommunicationSend(byte[] msg, String host, int port) throws IOException;


    public byte[] send(byte[] message) throws ExecutionException, InterruptedException {
        TaskState taskState = new TaskState();
        taskState.setType(MessageType.TYPE2);
        taskState.setWaitingTimeout(true);
        return send(message, taskState);
    }

    public byte[] send3(byte[] message) throws ExecutionException, InterruptedException {
        TaskState taskState = new TaskState();
        taskState.setType(MessageType.TYPE2);
        taskState.setWaitingTimeout(true);
        Future<byte[]> future = doSend(message, taskState);

        return future.get();
    }

    /**
     * 优先发送队列
     *
     * @param message
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public byte[] send2(byte[] message) throws ExecutionException, InterruptedException {
        TaskState taskState = new TaskState();
        taskState.setType(MessageType.TYPE1);
        taskState.setWaitingTimeout(false);
        Future<byte[]> future = doSend(message, taskState);

        return future.get();
    }

    public byte[] send(byte[] message, TaskState taskState) throws ExecutionException, InterruptedException {
        Future<byte[]> future = doSend(message, taskState);

        return future.get();
    }

    private Future<byte[]> doSend(byte[] message, TaskState taskState/*, String host, int port*/) {
        CompletableFuture<byte[]> future = new CompletableFuture<>();

        // 将消息和future包装成一个任务，记录入队时间
        MessageTask task = new MessageTask(message, future);
        //task.setHost(host);
        //task.setPort(port);
        task.setSendTimeout(responseTimeout);

        try {
            task.setState(taskState);
            put(task);
            // 启动超时检查线程
            scheduleTimeoutCheck(task, taskState.getType() == MessageType.TYPE1 ? messageQueue : messageQueue2);
        } catch (InterruptedException e) {
            future.completeExceptionally(e);
        }

        return future;
    }

    /**
     * 接收消息
     *
     * @throws Exception
     */
    protected void received() {

        while (isRun) {
            System.out.println("received");
            try {
                // 准备接收响应
                byte[] responseData = doReceive();
                if (receiveListener != null) {
                    receiveListener.receive(responseData, this);
                }
                if (task != null && !task.isCompleted()) {
                    task.revc = responseData;
                    task.receiveFuture.complete(responseData);
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }

    public abstract byte[] doReceive() throws Exception;

    public void put(MessageTask task) throws InterruptedException {
        lock.lock();
        try {
            if (task.getState().getType() == MessageType.TYPE1) {
                messageQueue.offer(task);
            } else {
                messageQueue2.offer(task);
            }
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }


    public MessageTask take() throws InterruptedException {
        lock.lock();
        try {
            while (messageQueue.isEmpty() && messageQueue2.isEmpty()) {
                notEmpty.await();
            }

            if (!messageQueue2.isEmpty()) {
                return messageQueue2.poll();
            }
            return messageQueue.poll();
        } finally {
            lock.unlock();
        }
    }


    /**
     * 调度超时检查
     */
    private void scheduleTimeoutCheck(MessageTask task, Queue<MessageTask> messageQueue) {
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.schedule(() -> {
            if (!task.isSent() && !task.isCompleted() && !task.getState().isWaitingTimeout()) {
                task.future.completeExceptionally(new QueueWaitingException("Message timed out in queue before sending " + queueTimeout));
                // 从队列中移除（如果还在队列中）
                messageQueue.remove(task);
            }
            scheduler.shutdownNow();
        }, queueTimeout, TimeUnit.MILLISECONDS);
    }


    public void close() throws IOException {
        executor.shutdownNow();
        INSTANCE = null;
        isRun = false;
    }

}
